package com.heima.kafka.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @author: wxw
 * @ClassName: ConsumerListener
 * @JdkVersion: JDK1.8
 * @Version: 1.0.0
 * @Since: 2025/2/21 14:28
 * @Description: 测试消费者
 */

@Component
public class ConsumerListener {


    // 单机
    //@KafkaListener(topics = "testTopic")
    // 集群 指定同一个分区   分区号
    @KafkaListener(topicPartitions = {@TopicPartition(topic = "testTopic", partitions = {"0"})})
    //@KafkaListener(topicPartitions = {@TopicPartition(topic = "testTopic", partitions = {"0","1","2"})})
    public void msg(ConsumerRecord<String, String> consumerRecord) {

        // 需要判断consumerRecord 是否为空

        // 方法1  与方法2效果一致
        //if (consumerRecord != null) {
        //    System.out.println(consumerRecord.key() + "==========" + consumerRecord.value());
        //}

        // 方法2
        Optional<ConsumerRecord<String, String>> optional = Optional.ofNullable(consumerRecord);
        optional.ifPresent(x -> System.out.println(x.key() + "=====" + x.value() + "=====" + x.partition() + "=====" + x.offset()));


    }
}
